package profile.cmlog

import org.apache.commons.lang3.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.elasticsearch.spark.rdd.EsSpark
import org.elasticsearch.spark.sql.EsSparkSQL

/**
  * Created by hunter.coder 涛哥  
  * 2019/4/18 11:08
  * 交流qq:657270652
  * Version: 1.0
  * 更多学习资料：https://blog.csdn.net/coderblack/
  * Description:  从增强后的运营商流量日志数据中抽取商品偏好关键字标签
  * 总流程：
  * 1.关联url内容知识库，获取url所对应的各类”知识“（1级品类名称，2级品类名称，3级品类名称，商品关键字）
  * 2.按dsp画像标签生成的流程，来计算标签即可
  *     -- 抽取需要的字段      ids（phone，imei，mac）
  *                         设备属性：操作系统名称，手机型号，屏幕分辨率.....
  *                         商品偏好关键词：品类1，品类2，品类3，商品关键字
  *     -- 将抽取出来的字段，整成画像标签数据结构：
  *                         每个标签模块： Map[String,List[(标签,分数)]]
  *                         各模块加一起： CmlogUserTagBean
  *
  *
  * 技术关切点：用何种方式去查询匹配url所对应的的内容知识
  * 方式1： 使用es客户端连接，在rdd的map算子中去逐条去es请求查询
  * 方式2： 直接将es知识库加载为dataframe，然后通过join方式来获取知识数据
  *
  * 方式1，速度慢
  * 方式2，速度快，但如果es知识库数据量庞大，则产生的rdd很大，从而可能耗费更多的内存
  *
  *
  * 注意：本代码中，缺失了查询统一图id的步骤，是用phone来替代gid写的
  *
  **/
object CmlogUserTagGener {

  def main(args: Array[String]): Unit = {

    var inpath = "G:\\testdata\\usertags\\cmlog\\b.dat"
    val outpath = "G:\\testdata\\usertags\\out-cmlogtags-T"

    val indexsource = "url_content/phone"

    if (args.size > 0) inpath = args(0)


    val spark = SparkSession
      .builder()
      .config("es.nodes", "c701:9200") //设置es.nodes
      .config("pushdown", "true") //执行sql语句时在elasticsearch中执行只返回需要的数据。这个参数在查询时设置比较有用
      .config("es.index.auto.create", "false") //如果为true，则会自动创建不存在的index
      .config("es.nodes.wan.only", "true")
      .master("local")
      .appName("移动流量日志标签生成")
      .getOrCreate()

    import spark.implicits._

    // 读取运营商日志
    val ds = spark.read.textFile(inpath)
    // 转换成dataframe
    val df = ds
      .map(s => {
        val split = s.split(",", -1)
        // idsTags
        val phone = split(0)
        val imei = split(1)
        // deviceTags
        val os_name = split(2)
        val device_type = split(3)

        val url = split(4)

        (url, phone, imei, os_name, device_type)
      })
      .toDF("url", "phone", "imei", "os_name", "device_type")

    // 加载es中的url内容知识库方式1：加载为RDD
    val esRdd: RDD[(String, collection.Map[String, AnyRef])] =
      EsSpark
        .esRDD(spark.sparkContext, indexsource)
        .map(tp => {  // tp: (docid,Map[String,AnyRef])
          val url: String = tp._2.getOrElse("url", "").asInstanceOf[String]
          (url, tp._2)
        })
        .filter(x => StringUtils.isNotBlank(x._1))

    // 加载es中的url内容知识库方式2：加载为DataFrame
    val esDf = EsSparkSQL.esDF(spark, indexsource)

    // 两份数据注册为临时表
    df.createTempView("df")
    esDf.createTempView("esdf")

    // 关联运营商流量日志数据和url知识库，关联不上的则抛弃
    val cmlogDf = spark.sql(
      """
        |select
        |a.*,b.*
        |from df a
        |join
        |esdf b
        |on a.url=concat('http:',b.url)
      """.stripMargin)

    cmlogDf.rdd.map(row => {
      // 抽取、封装设备属性标签map
      val os_name = row.getAs[String]("os_name")
      val device_type = row.getAs[String]("device_type")
      val deviceTags = Map("osname" -> List((os_name, 1d))) ++ Map("device_type" -> List((device_type, 1d)))

      // 抽取、封装ids标签map
      val imei = row.getAs[String]("imei")
      val phone = row.getAs[String]("phone")
      val idsTags = Map("imei" -> List((imei, 1d)), "phone" -> List((phone, 1d)))

      // 抽取、封装商品关键字偏好标签map
      val cat1 = row.getAs[String]("cat1")
      val cat2 = row.getAs[String]("cat2")
      val cat3 = row.getAs[String]("cat3")
      val kwds = row.getAs[String]("kwds")
      val goodsKwTags = Map("cat1" -> List((cat1, 1d))) ++ Map("cat2" -> List((cat2, 1d))) ++ Map("cat3" -> List((cat3, 1d))) ++ Map("kwds" -> kwds.split(" ").map((_, 1d)).toList)

      // 封装各模块标签到bean结构中
      CmlogUserTagBean(phone, idsTags, deviceTags, goodsKwTags)
    })
      //对相同设备的标签进行聚合
      .map(bean => (bean.phone, bean))
      .reduceByKey((b1, b2) => merge(b1, b2))
      .map(_._2)
      // 重构添加：将Map[String,List[(String,Double)]] 转换成 Map[String,Map[String,Double]]结构
      .map(b =>
        CmlogUserTagBean2(
          b.phone,
          b.idsTags.mapValues(_.toMap),
          b.deviceTags.mapValues(_.toMap),
          b.goodsKwTags.mapValues(_.toMap)
        )
      )
      .toDS()
      .coalesce(1)
      .write.parquet(outpath)


    spark.close()
  }


  /**
    * 标签聚合函数
    *
    * @param b1
    * @param b2
    * @return
    */
  def merge(b1: CmlogUserTagBean, b2: CmlogUserTagBean): CmlogUserTagBean = {

    val ids: Map[String, List[(String, Double)]] = mergeIds(b1.idsTags, b2.idsTags)

    val deviceTags: Map[String, List[(String, Double)]] = mergeDeviceTags(b1.deviceTags, b2.deviceTags)

    val goodsKwTags: Map[String, List[(String, Double)]] = mergeGoodsKwTags(b1.goodsKwTags, b2.goodsKwTags)

    CmlogUserTagBean(b1.phone, ids, deviceTags, goodsKwTags)
  }


  /**
    * ids聚合函数
    *
    * @param idsMap1
    * @param idsMap2
    * @return
    */
  def mergeIds(idsMap1: Map[String, List[(String, Double)]], idsMap2: Map[String, List[(String, Double)]]): Map[String, List[(String, Double)]] = {

    val imeiLst = idsMap1.getOrElse("imei", Nil) ++ idsMap2.getOrElse("imei", Nil)
    val imeiRes = imeiLst.groupBy(_._1).mapValues(_.map(_._2).sum).toList

    val phoneLst = idsMap1.getOrElse("phone", Nil) ++ idsMap2.getOrElse("phone", Nil)
    val phoneRes = phoneLst.groupBy(_._1).mapValues(_.map(_._2).sum).toList
    Map("imei" -> imeiRes, "phone" -> phoneRes)

  }

  /**
    * 设备信息标签聚合函数
    *
    * @param deviceTags1
    * @param deviceTags2
    * @return
    */
  def mergeDeviceTags(deviceTags1: Map[String, List[(String, Double)]], deviceTags2: Map[String, List[(String, Double)]]): Map[String, List[(String, Double)]] = {
    val osLst = deviceTags1.getOrElse("osname", Nil) ++ deviceTags2.getOrElse("osname", Nil)
    val osRes = osLst.groupBy(_._1).mapValues(_.map(_._2).sum).toList

    val dvcLst = deviceTags1.getOrElse("device_type", Nil) ++ deviceTags2.getOrElse("device_type", Nil)
    val dvcRes = dvcLst.groupBy(_._1).mapValues(_.map(_._2).sum).toList

    Map("osname" -> osRes, "device_type" -> dvcRes)
  }

  /**
    * 商品偏好关键字标签聚合函数
    *
    * @param kwds1
    * @param kwds2
    * @return
    */
  def mergeGoodsKwTags(kwds1: Map[String, List[(String, Double)]], kwds2: Map[String, List[(String, Double)]]): Map[String, List[(String, Double)]] = {

    // 将两个bean中的 interest关键词，加到一个列表里
    val kwdsTags: List[(String, Double)] = kwds1.getOrElse("kwds", Nil) ++ kwds2.getOrElse("kwds", Nil)

    // 将相同关键词的分数累加
    val kwdsTagsRes = kwdsTags.groupBy(_._1).mapValues(_.map(_._2).sum).toList

    // 将两个bean中的 cat1，加到一个列表里
    val cat1Lst: List[(String, Double)] = kwds1.getOrElse("cat1", Nil) ++ kwds2.getOrElse("cat1", Nil)
    // 将相同关键词的分数累加
    val cat1Res = cat1Lst.groupBy(_._1).mapValues(_.map(_._2).sum).toList

    // 将两个bean中的 cat2，加到一个列表里
    val cat2Lst: List[(String, Double)] = kwds1.getOrElse("cat2", Nil) ++ kwds2.getOrElse("cat2", Nil)
    // 将相同关键词的分数累加
    val cat2Res = cat2Lst.groupBy(_._1).mapValues(_.map(_._2).sum).toList

    // 将两个bean中的 cat3，加到一个列表里
    val cat3Lst: List[(String, Double)] = kwds1.getOrElse("cat3", Nil) ++ kwds2.getOrElse("cat3", Nil)
    // 将相同关键词的分数累加
    val cat3Res = cat3Lst.groupBy(_._1).mapValues(_.map(_._2).sum).toList

    Map("kwds" -> kwdsTagsRes, "cat1" -> cat1Res, "cat2" -> cat2Res, "cat3" -> cat3Res)
  }

}
